package com.github.facadefs.hdfs;

import com.github.facadefs.FileView;
import com.github.facadefs.support.SessionBasedFileSystem;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import com.github.facadefs.Selector;
import com.github.facadefs.util.Paths;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

/**
 *
 */
public class HdfsFileSystem extends SessionBasedFileSystem<FileSystem> {
    private final URI hdfsUri;

    public HdfsFileSystem(final URI hdfsUri) {
        this.hdfsUri = hdfsUri;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean exists(final String path) throws IOException {
        final String absPath = toAbsolutePath(path);
        final FileSystem fs = connect();
        try {
            return fs.exists(new Path(absPath));
        } finally {
            disconnect(fs);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected FileSystem connect() throws IOException {
        final Configuration conf = new Configuration();
        return org.apache.hadoop.fs.FileSystem.get(hdfsUri, conf);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void disconnect(final FileSystem fs) {
        try {
            fs.close();
        } catch (final IOException e) {
            // ignore
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected FileView stat(final FileSystem fs, final String path) throws IOException {
        final String absPath = toAbsolutePath(path);
        try {
            final FileStatus stat = stat0(fs, absPath);
            return null != stat ? toView(stat) : null;
        } catch (final FileNotFoundException n) {
            return null;
        }
    }

    private FileStatus stat0(final FileSystem fs, final String absPath) throws IOException {
        try {
            return fs.getFileStatus(new Path(absPath));
        } catch (final FileNotFoundException n) {
            return null;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected FileView[] ls(final FileSystem fs, final String path) throws IOException {
        final String absPath = toAbsolutePath(path);
        final List<FileView> views = new ArrayList<FileView>();
        final RemoteIterator<LocatedFileStatus> it = fs.listFiles(new Path(absPath), false);
        while (it.hasNext()) {
            final FileView view = toView(it.next());
            if (null != view) {
                views.add(view);
            }
        }
        return views.toArray(new FileView[0]);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void ls(final FileSystem fs, final String path, final Selector selector) throws IOException {
        final String absPath = toAbsolutePath(path);
        final RemoteIterator<LocatedFileStatus> it = fs.listFiles(new Path(absPath), false);
        while (it.hasNext()) {
            final FileView view = toView(it.next());
            if (Selector.CONTINUE != selector.select(view)) {
                break;
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected boolean mkdirs(final FileSystem fs, final String path) throws IOException {
        final String absPath = toAbsolutePath(path);
        return fs.mkdirs(new Path(absPath));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected InputStream open(final FileSystem fs, final String path) throws IOException {
        final String absPath = toAbsolutePath(path);
        return fs.open(new Path(absPath));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected OutputStream write(final FileSystem fs, final String path, final String contentType, final boolean override) throws IOException {
        final String absPath = toAbsolutePath(path);
        return fs.create(new Path(absPath), override);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void rm(final FileSystem fs, final String path, final boolean recurse) throws IOException {
        final String absPath = toAbsolutePath(path);
        fs.delete(new Path(absPath), recurse);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void rename(final FileSystem fs, final String oldPath, final String newPath) throws IOException {
        final String absOldPath = toAbsolutePath(oldPath);
        final String absNewPath = toAbsolutePath(newPath);
        if (!fs.rename(new Path(absOldPath), new Path(absNewPath))) {
            throw new IOException("rename failed: " + oldPath + " -> " + newPath);
        }
    }

    private String toAbsolutePath(final String path) {
        return Paths.toAbsolutePath(hdfsUri.getPath(), path);
    }

    private String toRelativePath(final String path) {
        return Paths.toRelativePath(hdfsUri.getPath(), path);
    }

    private FileView toView(final FileStatus st) {
        if (null == st) {
            return null;
        }

        final String path = toRelativePath(st.getPath().toUri().getPath());
        final long length = st.getLen();
        final boolean isDirectory = st.isDirectory();
        final long mtime = st.getModificationTime();
        final long atime = st.getAccessTime();
        final String owner = st.getOwner();
        final String group = st.getGroup();
        final FsPermission perm = st.getPermission();

        String link = null;
        try {
            final Path symlink = st.getSymlink();
            link = null != symlink ? symlink.toUri().getPath() : null;
        } catch (final IOException e) {
            // ignore
        }

        return new FileView(path, length, isDirectory, mtime, atime, perm.toOctal(), owner, group, link);
    }

}
